package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeAddCart {
    public static void main(String[] args) {
        // TODO: 2022/5/25  1.获取流执行环境和表执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //设置状态过期时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        // TODO: 2022/5/25 2.检查点相关设置



        // TODO: 2022/5/25 3.从kafka 主题 topic_log读取数据创建动态表
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` STRING,\n" +
                "  `table` STRING,\n" +
                "  `type` STRING,\n" +
                "  `ts` STRING,\n" +
                "  `proc_time` as proctime(),\n" +
                "  `data` MAP<STRING,STRING>,\n" +
                "  `old` MAP<STRING,STRING>\n" +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db","dwd_trade_cart_add_group"));

        //tableEnv.executeSql("select * from topic_db").print();





        // TODO: 2022/5/25 4.从上边创建的动态表过滤出加购数据---生成动态表
        Table cartAdd = tableEnv.sqlQuery("select\n" +
                "    data['id'] id,\n" +
                "    data['user_id'] user_id,\n" +
                "    data['sku_id' ] sku_id,  \n" +
                "    data['source_type'] source_type,\n" +
                "    data['source_id'] source_id,\n" +
                "    IF(`type`= 'insert',\n" +
                "      data['sku_num'],\n" +
                "      cast((cast(data['sku_num'] as int) - cast(`old`['sku_num'] as int)) as string)\n" +
                "    ) sku_num,\n" +
                "    ts,\n" +
                "    proc_time\n" +
                "from topic_db\n" +
                "where\n" +
                "    `table` = 'cart_info'\n" +
                "and\n" +
                "    (\n" +
                "        `type` = 'insert'\n" +
                "      or\n" +
                "        (`type` = 'update' and `old`['sku_num'] is not null and cast(data['sku_num'] as int) >  cast(`old`['sku_num'] as int))\n" +
                "    )");

        tableEnv.createTemporaryView("cart_add",cartAdd);

        //tableEnv.executeSql("select * from cart_add").print();

        // TODO: 2022/5/25 5.从mysql数据库读取字典数据表---动态字典表
        tableEnv.executeSql(MysqlUtil.getBaseDicDDL());

        //tableEnv.executeSql("select * from base_dic").print();
        // TODO: 2022/5/25  6.关联两张动态表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "    id,user_id,sku_id,source_id,source_type,dic_name as source_type_name,sku_num,ts\n" +
                "from\n" +
                "    cart_add cadd\n" +
                "join\n" +
                "    base_dic  FOR SYSTEM_TIME AS OF cadd.proc_time dic\n" +
                "on \n" +
                "    cadd.source_type = dic.dic_code");

        tableEnv.createTemporaryView("res_table",resultTable);
        //tableEnv.executeSql("select * from res_table").print();


        // TODO: 2022/5/25 7. 创建动态表  映射关联kafka的主题 dwd_trade_cart_add 字段和查询出来的字段类型，字段一一对应
        tableEnv.executeSql("CREATE TABLE dwd_trade_cart_add (\n" +
                "  id string,\n" +
                "  user_id string,\n" +
                "  sku_id string,\n" +
                "  source_id string,\n" +
                "  source_type string,\n" +
                "  source_type_name string,\n" +
                "  sku_num string,\n" +
                "  ts string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_cart_add"));


        // TODO: 2022/5/25 8.将关联的数据写入到映射kafka主题的表
        tableEnv.executeSql("insert into dwd_trade_cart_add select * from res_table");

    }
}
